package cn.dmp.charts.terminalAndDevice

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

object NetworkMannerDistributionV1 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("NetworkTypeDistributionV1").setMaster("local[3]")
    val sc: SparkContext = new SparkContext(conf)
    //包装sparkcontext后增强
    val sqlContext= new SQLContext(sc)

    val logDf: DataFrame = sqlContext.read.load(args(0))
    //选出字段
    val select: DataFrame = logDf.select("networkmannername","requestmode","processnode","iseffective","isbilling","isbid","iswin","adorderid","winprice","adpayment")
    //聚合
    select.registerTempTable("t_distribution")
    val sql = "select networkmannername," +
      " sum(case when (requestmode =1 and processnode>=1) then 1 else 0 end) primaryRequest," +
      " sum(case when (requestmode =1 and processnode>=2) then 1 else 0 end ) effectiveRequest," +
      " sum(case when (requestmode =1 and processnode>=3) then 1 else 0 end ) AdRequest, " +
      " sum(case when (iseffective = 1 and isbilling = 1 and isbid = 1 and adorderid !=0) then 1 else 0 end) biddingTimes," +
      " sum(case when (iseffective = 1 and isbilling = 1 and iswin = 1 ) then 1 else 0 end) succesedBiddingTimes, " +
      " sum(case when (requestmode =2 and iseffective =1 ) then 1 else 0 end ) showTimes, " +
      " sum(case when (requestmode =3 and iseffective =1 ) then 1 else 0 end ) clickTimes, " +
      " sum(if((iseffective=1 and isbilling=1 and iswin =1),winprice,0))/1000 AdConsume, " +
      " sum(if((iseffective=1 and isbilling=1 and iswin =1),adpayment,0))/1000 AdCost " +
      "from t_distribution" +
      " group by networkmannername "

    val props = new Properties()
    props.put("user","root")
    props.put("password","123456")
    sqlContext.sql(sql).write.mode("append").jdbc("jdbc:mysql://localhost:3306/npm?useUnicode=true&characterEncoding=utf-8", "NetworkTypeDistributionV1", props)




    sc.stop()

  }
}
